package com.gome.ocean.controller.datax;

import java.io.IOException;
import java.util.Date;
import java.util.Map;

import org.apache.commons.lang.StringUtils;
import org.quartz.SchedulerException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.ResponseBody;

import com.alibaba.fastjson.JSON;
import com.gome.ocean.common.constants.DataxContants;
import com.gome.ocean.common.enums.State;
import com.gome.ocean.common.exception.JobServiceException;
import com.gome.ocean.common.utils.IOUtils;
import com.gome.ocean.dao.model.datasource.DatasourceBO;
import com.gome.ocean.dao.model.datax.JobBO;
import com.gome.ocean.dao.model.datax.JobHistoryBO;
import com.gome.ocean.dao.model.yugong.YgManagerBO;
import com.gome.ocean.dao.model.yugong.YgTaskBO;
import com.gome.ocean.service.datax.DatasourceService;
import com.gome.ocean.service.datax.JobHistoryService;
import com.gome.ocean.service.datax.JobService;
import com.gome.ocean.service.task.CallBack;
import com.gome.ocean.service.thread.ThreadFactory;
import com.gome.ocean.service.utils.CmdToolkit;
import com.gome.ocean.service.velocity.VelocityBean;
import com.gome.ocean.service.yugong.YgManagerService;
import com.gome.ocean.service.yugong.YgTaskService;
import com.google.common.collect.Maps;

@Controller
public class DataxApiController {

    private static final Logger LOG                  = LoggerFactory.getLogger(DataxApiController.class);

    private static final String RESULT_MAP_ERROR_MSG = "errorMsg";
    @Autowired
    private JobService          jobService;
    @Autowired
    private VelocityBean        velocityBean;
    @Autowired
    private ThreadFactory       threadFactory;
    @Autowired
    private JobHistoryService   jobHistoryService;
    @Autowired
    private YgTaskService       ygTaskService;
    @Autowired
    private YgManagerService    ygManagerService;
    @Autowired
    private DatasourceService   datasourceService;

    @RequestMapping(value = "/api/context", method = RequestMethod.GET)
    @ResponseBody
    public String getJonContext(Long jobId) {
        String result = "{'result':'-1'}";
        if (null == jobId || jobId.intValue() < 0) {
            LOG.info("job id is not null! please check!");
            return result; // datax规定 不需修改
        }
        JobBO jobBO = jobService.findJob(jobId);
        if (jobBO == null) {
            LOG.info("jobBO is null!");
            return result;
        }
        if (StringUtils.isEmpty(jobBO.getContext())) {
            LOG.info("jobBO context is null!");
            return result;
        }
        return velocityBean.processJobContext(jobBO.getContext());
    }

    @RequestMapping(value = "/api/yugong/start", method = RequestMethod.GET)
    @ResponseBody
    public String YugongStart(Long taskId) {
        String result = "{'result':'-1'}";
        if (null == taskId || taskId.intValue() < 0) {
            LOG.info("taskId id is not null! please check!");
            return result; // datax规定 不需修改
        }
        YgTaskBO ygTask = ygTaskService.findYgTaskBOById(taskId);
        if (ygTask == null) {
            LOG.error("ygTaskBO is null!");
            return result;
        }
        YgManagerBO ygManager = ygManagerService.findYgManagerBOById(ygTask.getManagerId());
        if (ygManager == null) {
            LOG.error("yugong manager is error!managerId:" + ygTask.getManagerId());
            return result;
        }
        if (ygManager.getIsDisabled() == 1) {
            LOG.error("yugong manager is disable!please check!" + ygTask.getManagerId());
            return result;
        }
        if (ygManager.getIsOccupy() == 1) {
            LOG.error("yugong manager is occupy!please check!" + ygTask.getManagerId());
            return result;
        }
        return ygTaskService.startTask(ygManager, ygTask);
    }

    @RequestMapping(value = "/api/job/interruptJob", method = RequestMethod.GET)
    @ResponseBody
    public String interruptJob(Long id) {
        String result = "{'result':'-1'}";
        if (null == id || id.intValue() < 0) {
            LOG.info("jod id is not null! please check!");
            return result; // datax规定 不需修改
        }
        final JobBO job = jobService.findJob(id);
        if (job == null) {
            LOG.info("ygTaskBO is null!");
            return result;
        }
        final JobHistoryBO jobHistoryBO = jobHistoryService.findCurrentJob(job.getId());
        if (jobHistoryBO == null) {
            LOG.info("jobHistoryBO is null!");
            return result;
        }
        if (job.isCronJob()) {
            try {
                jobService.interrupt(job);
            } catch (SchedulerException | JobServiceException e) {
                LOG.info("jobBO stop is error!", e);
                return result;
            }
        } else {
            CmdToolkit.executeCmd(DataxContants.getKillTaskCommand(jobHistoryBO.getTJobId()), new CallBack() {

                @Override
                public void callBack() {
                    threadFactory.stop(jobHistoryBO.getTJobId());

                    job.setState(State.FAIL.getValue());
                    job.setEndTime(new Date());

                    jobHistoryBO.setState(State.FAIL.getValue());
                    jobHistoryBO.setSubmitUser(job.getSubmitUser());
                    jobHistoryBO.setSubmitTime(job.getSubmitTime());
                    jobHistoryBO.setSubmitIp(job.getSubmitIp());
                    jobHistoryBO.setPipelineName(job.getPipelineName());
                    jobHistoryBO.setCronExpression(job.getCronExpression());

                    jobHistoryBO.setStartTime(job.getStartTime());
                    jobHistoryBO.setEndTime(job.getEndTime());
                    jobHistoryBO.setSpeedBytes("0");
                    jobHistoryBO.setSpeedRecords(0l);
                    jobHistoryBO.setTotalRecords(0l);
                    jobHistoryBO.setErrorRecords(0l);
                    jobService.updateJobBO(job);
                    jobHistoryService.updateJobHistoryById(jobHistoryBO);
                }
            });
            return "{'result':'ok'}";
        }
        return result;
    }

    @RequestMapping(value = "/api/yugong/stop", method = RequestMethod.GET)
    @ResponseBody
    public String YugongStop(Long taskId) {
        String result = "{'result':'-1'}";
        if (null == taskId || taskId.intValue() < 0) {
            LOG.info("taskId id is not null! please check!");
            return result; // datax规定 不需修改
        }
        YgTaskBO ygTask = ygTaskService.findYgTaskBOById(taskId);
        if (ygTask == null) {
            LOG.error("ygTaskBO is null!");
            return result;
        }
        YgManagerBO ygManager = ygManagerService.findYgManagerBOById(ygTask.getManagerId());
        if (ygManager == null) {
            LOG.error("yugong manager is error!managerId:" + ygTask.getManagerId());
            return result;
        }
        if (ygManager.getIsDisabled() == 1) {
            LOG.error("yugong manager is disable!please check!" + ygTask.getManagerId());
            return result;
        }
        if (ygManager.getIsOccupy() == 0) {
            LOG.error("yugong manager is not occupy!please check!" + ygTask.getManagerId());
            return result;
        }
        return ygTaskService.stopTask(ygManager, ygTask);
    }

    @RequestMapping(value = "/api/yugong/log", method = RequestMethod.GET)
    @ResponseBody
    public String getYugongLog(Long id, String callback, Long beginIndex) {
        Map<String, Object> resultMap = Maps.newHashMap();
        resultMap.put(IOUtils.DEFAULT_LOG_BEGININDEX, -1);
        resultMap.put(RESULT_MAP_ERROR_MSG, "");
        if (null == id || id.intValue() < 0) {
            LOG.info("YgTaskBO id is not null! please check!");
            resultMap.put(RESULT_MAP_ERROR_MSG, "您没有输入Id参数，或者Id参数值不正确");
            return toJsonP(resultMap, callback); // datax规定 不需修改
        }
        YgTaskBO ygTaskBO = ygTaskService.findYgTaskBOById(id);
        if (ygTaskBO == null) {
            LOG.info("ygTaskBO is null!");
            resultMap.put(RESULT_MAP_ERROR_MSG, "未查找到相关的ygTaskBO记录");
            return toJsonP(resultMap, callback);
        }
        YgManagerBO ygManagerBO = ygManagerService.findYgManagerBOById(ygTaskBO.getManagerId());
        if (null == ygManagerBO) {
            LOG.info("ygManagerBO is null!");
            resultMap.put(RESULT_MAP_ERROR_MSG, "未查找到相关的ygManagerBO记录");
            return toJsonP(resultMap, callback);
        }
        String ygPath = ygManagerBO.getYgLocation();
        if (StringUtils.isEmpty(ygPath)) {
            LOG.info("ygPath is null!");
            resultMap.put(RESULT_MAP_ERROR_MSG, "读取日志文件过程中发生异常，请联系管理员");
            return toJsonP(resultMap, callback);
        }
        DatasourceBO appDatasourceBO = datasourceService.findDatasourceBOById(ygTaskBO.getSourceDatasourceId());
        if (appDatasourceBO == null) {
            LOG.info("appDatasourceBO is null!");
            resultMap.put(RESULT_MAP_ERROR_MSG, "读取日志文件过程中发生异常，请联系管理员");
            return toJsonP(resultMap, callback);
        }
        try {
            StringBuffer logFileUrl = new StringBuffer(ygPath);
            logFileUrl.append("/logs/")
                .append(appDatasourceBO.getUserName())
                .append(".")
                .append(ygTaskBO.getTableName())
                .append("/")
                .append("table.log");
            LOG.info("读取日志文件，地址：" + logFileUrl.toString());
            Map<String, Object> resultMapTemp = IOUtils.readFileByRandomAccess(logFileUrl.toString(), beginIndex);
            resultMap.putAll(resultMapTemp);
            return toJsonP(resultMap, callback);
        } catch (IOException e) {
            LOG.error(e.getMessage(), e);
            resultMap.put(RESULT_MAP_ERROR_MSG, "读取日志文件过程中发生异常，请联系管理员");
            return toJsonP(resultMap, callback);
        }
    }

    /**
     * 查看datax运行日志<br>
     * 支持jsonP跨域访问
     *
     * @param historyJobId
     * @param beginIndex
     * @param callback
     * @return
     */
    @RequestMapping(value = "/api/log", method = RequestMethod.GET)
    @ResponseBody
    public String getJonHisLogInfo(Long historyJobId, Long beginIndex, String callback) {
        Map<String, Object> resultMap = Maps.newHashMap();
        resultMap.put(IOUtils.DEFAULT_LOG_BEGININDEX, -1);
        resultMap.put(RESULT_MAP_ERROR_MSG, "");
        if (null == historyJobId || historyJobId.intValue() < 0) {
            LOG.info("jobHistoryBO id is not null! please check!");
            resultMap.put(RESULT_MAP_ERROR_MSG, "您没有输入historyJobId参数，或者historyJobId参数值不正确");
            return toJsonP(resultMap, callback); // datax规定 不需修改
        }
        JobHistoryBO jobHistoryBO = jobHistoryService.findJobHistoryById(historyJobId);
        if (jobHistoryBO == null) {
            LOG.info("jobHistoryBO is null!");
            resultMap.put(RESULT_MAP_ERROR_MSG, "未查找到相关的jobHistory记录");
            return toJsonP(resultMap, callback);
        }
        if (StringUtils.isEmpty(jobHistoryBO.getLogUrl())) {
            LOG.info("jobHistoryBO logUrl is null!");
            resultMap.put(RESULT_MAP_ERROR_MSG, "读取日志文件过程中发生异常，请联系管理员");
            return toJsonP(resultMap, callback);
        }
        try {
            Map<String, Object> resultMapTemp = IOUtils.readFileByRandomAccess(jobHistoryBO.getLogUrl(), beginIndex);
            resultMap.putAll(resultMapTemp);
            return toJsonP(resultMap, callback);
        } catch (IOException e) {
            LOG.error("读取日志文件出错！logUrl:" + jobHistoryBO.getLogUrl());
            resultMap.put(RESULT_MAP_ERROR_MSG, "读取日志文件过程中发生异常，请联系管理员");
            return toJsonP(resultMap, callback);
        }
    }

    /**
     * jsonp格式
     *
     * @param obj
     * @param callback
     * @return
     */
    private String toJsonP(Object obj, String callback) {
        String json = JSON.toJSONString(obj);
        return callback.concat("(").concat(json).concat(")");
    }
}
